package consumer_manager

import (
	"fmt"
	"time"

	"gitee.com/Sxiaobai/gs/gsdb"
	"gitee.com/Sxiaobai/gs/gstool"
	"github.com/spf13/cast"
)

// DatabaseManager 数据库管理器
type DatabaseManager struct {
	db        *gsdb.GsMysql
	tableName string
	logger    *gstool.GsSlog
}

// NewDatabaseManager 创建数据库管理器
func NewDatabaseManager(db *gsdb.GsMysql, tableName string, logger *gstool.GsSlog) *DatabaseManager {
	return &DatabaseManager{
		db:        db,
		tableName: tableName,
		logger:    logger,
	}
}

// CreateTable 创建表
// func (dm *DatabaseManager) CreateTable() error {
// 	sql := fmt.Sprintf(`
// 		CREATE TABLE IF NOT EXISTS %s (
// `id` int unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
// `consumer_name` varchar(100) NOT NULL COMMENT '消费者名称',
// `server_ip` varchar(50) NOT NULL COMMENT '服务器内网IP',
// `port` varchar(10) NOT NULL COMMENT '端口号',
// `nsq_lookup_host` varchar(100) NOT NULL COMMENT 'NSQ Lookup地址',
// `topic` varchar(100) NOT NULL COMMENT 'NSQ Topic',
// `channel` varchar(100) NOT NULL COMMENT 'NSQ Channel',
// `max_consumer_count` int unsigned DEFAULT '0' COMMENT '最大消费者个数',
// `consumer_count` int unsigned NOT NULL DEFAULT '0' COMMENT '消费者个数',
// `create_time` int NOT NULL DEFAULT '0' COMMENT '创建时间',
// `update_time` int unsigned NOT NULL DEFAULT '0' COMMENT '更新时间',
// 			PRIMARY KEY (id),
// 			UNIQUE KEY uk_consumer_server_port_topic_channel (consumer_name, server_ip, port, topic, channel)
// 		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消费者监控表'
// 	`, dm.tableName)

// 	_, err := dm.db.ExecBySql(sql).Exec()
// 	if err != nil {
// 		dm.logger.Errof("创建表失败: %v", err)
// 		return err
// 	}

// 	dm.logger.Infof("表 %s 创建成功", dm.tableName)
// 	return nil
// }

// UpsertStatus 更新或插入状态（优化版本）
func (dm *DatabaseManager) UpsertStatus(consumerName, serverIP, port, nsqLookupHost, nsqHTTPAddress, nsqAdminAddress, topic, channel, consumerRemark string, maxConsumerCount, consumerCount, spawnConsumerCount, depthLimit int) error {
	now := time.Now().Unix()

	// 先尝试查询现有记录
	existingStatus, err := dm.GetStatus(consumerName, serverIP, port, topic, channel)
	if err != nil {
		// 记录不存在，执行插入
		insertSQL := fmt.Sprintf(`
			INSERT INTO %s (consumer_name, server_ip, port, nsq_lookup_host, nsq_http_address, nsq_admin_address, topic, channel, max_consumer_count, consumer_count, spawn_consumer_count, depth_limit, create_time, update_time, consumer_remark) 
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		`, dm.tableName)

		_, err = dm.db.ExecBySql(insertSQL, consumerName, serverIP, port, nsqLookupHost, nsqHTTPAddress, nsqAdminAddress, topic, channel, maxConsumerCount, consumerCount, spawnConsumerCount, depthLimit, now, now, consumerRemark).Exec()
		if err != nil {
			dm.logger.Errof("插入消费者状态失败: %v", err)
			return err
		}
		dm.logger.Debugf("消费者状态插入成功: %s@%s:%s [%s/%s/%s], max_count=%d, count=%d", consumerName, serverIP, port, nsqLookupHost, topic, channel, maxConsumerCount, consumerCount)
	} else {
		// 记录存在，执行更新
		updateSQL := fmt.Sprintf(`
			UPDATE %s 
			SET nsq_lookup_host=?, nsq_http_address=?, nsq_admin_address=?, max_consumer_count=?, consumer_count=?, spawn_consumer_count=?, depth_limit=?, update_time=?, consumer_remark=? 
			WHERE consumer_name=? AND server_ip=? AND port=? AND topic=? AND channel=?
		`, dm.tableName)

		_, err = dm.db.ExecBySql(updateSQL, nsqLookupHost, nsqHTTPAddress, nsqAdminAddress, maxConsumerCount, consumerCount, spawnConsumerCount, depthLimit, now, consumerRemark, consumerName, serverIP, port, topic, channel).Exec()
		if err != nil {
			dm.logger.Errof("更新消费者状态失败: %v", err)
			return err
		}
		dm.logger.Infof("消费者状态更新成功: %s@%s:%s [%s/%s/%s], max_count=%d, count=%d (ID: %d)", consumerName, serverIP, port, nsqLookupHost, topic, channel, maxConsumerCount, consumerCount, existingStatus.ID)
	}

	return nil
}

// GetAllStatus 获取所有状态
func (dm *DatabaseManager) GetAllStatus() ([]ConsumerStatus, error) {
	sql := fmt.Sprintf(`
		SELECT id, consumer_name, server_ip, port, nsq_lookup_host, nsq_http_address, nsq_admin_address, topic, channel, max_consumer_count, consumer_count, spawn_consumer_count, depth_limit, scaling_mode, create_time, update_time, consumer_remark 
		FROM %s 
		ORDER BY update_time DESC
	`, dm.tableName)

	result, err := dm.db.QueryBySql(sql).All()
	if err != nil {
		return nil, err
	}

	// 检查查询结果是否为空
	if len(result) == 0 {
		return nil, fmt.Errorf("未找到任何消费者状态记录")
	}

	var statuses []ConsumerStatus
	for _, row := range result {
		var status ConsumerStatus
		status.ID = cast.ToInt(row["id"])
		status.ConsumerName = cast.ToString(row["consumer_name"])
		status.ServerIP = cast.ToString(row["server_ip"])
		status.Port = cast.ToString(row["port"])
		status.NSQLookupHost = cast.ToString(row["nsq_lookup_host"])
		status.NSQHTTPAddress = cast.ToString(row["nsq_http_address"])
		status.NSQAdminAddress = cast.ToString(row["nsq_admin_address"])
		status.Topic = cast.ToString(row["topic"])
		status.Channel = cast.ToString(row["channel"])
		status.MaxConsumerCount = cast.ToInt(row["max_consumer_count"])
		status.ConsumerCount = cast.ToInt(row["consumer_count"])
		status.SpawnConsumerCount = cast.ToInt(row["spawn_consumer_count"])
		status.DepthLimit = cast.ToInt(row["depth_limit"])
		status.ScalingMode = cast.ToString(row["scaling_mode"])
		status.CreateTime = time.Unix(cast.ToInt64(row["create_time"]), 0)
		status.UpdateTime = time.Unix(cast.ToInt64(row["update_time"]), 0)
		status.ConsumerRemark = cast.ToString(row["consumer_remark"])
		statuses = append(statuses, status)
	}

	return statuses, nil
}

// GetStatus 获取状态
func (dm *DatabaseManager) GetStatus(consumerName, serverIP, port, topic, channel string) (*ConsumerStatus, error) {
	sql := fmt.Sprintf(`
		SELECT id, consumer_name, server_ip, port, nsq_lookup_host, nsq_http_address, nsq_admin_address, topic, channel, max_consumer_count, consumer_count, spawn_consumer_count, depth_limit, scaling_mode, create_time, update_time, consumer_remark 
		FROM %s 
		WHERE consumer_name = ? AND server_ip = ? AND port = ? AND topic = ? AND channel = ?
	`, dm.tableName)

	result, err := dm.db.QueryBySql(sql, consumerName, serverIP, port, topic, channel).One()
	if err != nil {
		return nil, err
	}

	// 检查查询结果是否为空map（根据One()方法的实现，空结果返回空map）
	if len(result) == 0 {
		return nil, fmt.Errorf("未找到指定的消费者状态记录: %s@%s:%s [%s/%s]", consumerName, serverIP, port, topic, channel)
	}

	var status ConsumerStatus
	status.ID = cast.ToInt(result["id"])
	status.ConsumerName = cast.ToString(result["consumer_name"])
	status.ServerIP = cast.ToString(result["server_ip"])
	status.Port = cast.ToString(result["port"])
	status.NSQLookupHost = cast.ToString(result["nsq_lookup_host"])
	status.NSQHTTPAddress = cast.ToString(result["nsq_http_address"])
	status.NSQAdminAddress = cast.ToString(result["nsq_admin_address"])
	status.Topic = cast.ToString(result["topic"])
	status.Channel = cast.ToString(result["channel"])
	status.MaxConsumerCount = cast.ToInt(result["max_consumer_count"])
	status.ConsumerCount = cast.ToInt(result["consumer_count"])
	status.SpawnConsumerCount = cast.ToInt(result["spawn_consumer_count"])
	status.DepthLimit = cast.ToInt(result["depth_limit"])
	status.ScalingMode = cast.ToString(result["scaling_mode"])
	status.CreateTime = time.Unix(cast.ToInt64(result["create_time"]), 0)
	status.UpdateTime = time.Unix(cast.ToInt64(result["update_time"]), 0)
	status.ConsumerRemark = cast.ToString(result["consumer_remark"])

	return &status, nil
}

// ToggleScalingMode 切换scaling_mode状态
func (dm *DatabaseManager) ToggleScalingMode(consumerName, serverIP, port, topic, channel string) error {
	// 先获取当前状态
	status, err := dm.GetStatus(consumerName, serverIP, port, topic, channel)
	if err != nil {
		return fmt.Errorf("获取消费者状态失败: %v", err)
	}

	// 切换scaling_mode状态
	newMode := "manual"
	if status.ScalingMode == "manual" {
		newMode = "auto"
	}

	// 更新数据库
	updateSQL := fmt.Sprintf(`
		UPDATE %s 
		SET scaling_mode=?, update_time=? 
		WHERE consumer_name=? AND server_ip=? AND port=? AND topic=? AND channel=?
	`, dm.tableName)

	now := time.Now().Unix()
	_, err = dm.db.ExecBySql(updateSQL, newMode, now, consumerName, serverIP, port, topic, channel).Exec()
	if err != nil {
		dm.logger.Errof("切换scaling_mode状态失败: %v", err)
		return err
	}

	dm.logger.Infof("scaling_mode状态切换成功: %s@%s:%s [%s/%s] %s -> %s", consumerName, serverIP, port, topic, channel, status.ScalingMode, newMode)
	return nil
}
